package org.example.develop;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import java.util.List;
import java.util.Optional;

/**
 * 定义一个类,用于装载计算好的 DataSet,并作为方法的参数返回.
 */


@Data
@NoArgsConstructor
@AllArgsConstructor
public class DatasetContainer {
    private Dataset<Row> dataset1; // 写入 ads_phm_fault 中的结果
    private Dataset<Row> dataset2; // 写入 t_phm_sqwx 中的结果
    private Dataset<Row> dataset3; // 用于健康评估的结果


    /***
     * 定义一个方法,用来处理 30个模型调用处理后的结果,也就是每个DatasetContainer中 DataSet1 union 结果 , DataSet2 union 结果 , DataSet3 union 结果.
     * @param containers 传入一个List<DatasetContainer>
     * @return 返回一个 DatasetContainer,里面存放的是30模型每个属性的union结果. 并实现最终入库操作.
     */
    public static DatasetContainer unionDatasets(List<DatasetContainer> containers) {
        Dataset<Row> resultDataset1 = null;
        Dataset<Row> resultDataset2 = null;
        Dataset<Row> resultDataset3 = null;

        for (DatasetContainer container : containers) {
            // 对dataset1进行union操作
            if (container.dataset1 != null && container.dataset1.count() > 0) {
                if (resultDataset1 == null) {
                    resultDataset1 = container.dataset1;
                } else {
                    resultDataset1 = resultDataset1.union(container.dataset1);
                }
            }

            // 对dataset2进行union操作
            if (container.dataset2 != null && container.dataset2.count() > 0) {
                if (resultDataset2 == null) {
                    resultDataset2 = container.dataset2;
                } else {
                    resultDataset2 = resultDataset2.union(container.dataset2);
                }
            }

            // 对dataset3进行union操作
            if (container.dataset3 != null && container.dataset3.count() > 0) {
                if (resultDataset3 == null) {
                    resultDataset3 = container.dataset3;
                } else {
                    resultDataset3 = resultDataset3.union(container.dataset3);
                }
            }
        }

        DatasetContainer resultContainer = new DatasetContainer();
        resultContainer.dataset1 = resultDataset1;
        resultContainer.dataset2 = resultDataset2;
        resultContainer.dataset3 = resultDataset3;

        return resultContainer;
    }

    public static DatasetContainer unionDatasets2(List<DatasetContainer> containers) {
        Dataset<Row> resultDataset1 = null;
        Dataset<Row> resultDataset2 = null;
        Dataset<Row> resultDataset3 = null;

        for (DatasetContainer container : containers) {
            // 对dataset1进行union操作
            if (container.dataset1 != null && !container.dataset1.toJavaRDD().isEmpty()) {
                if (resultDataset1 == null) {
                    resultDataset1 = container.dataset1;
                } else {
                    resultDataset1 = resultDataset1.union(container.dataset1);
                }
            }

            // 对dataset2进行union操作
            if (container.dataset2 != null && !container.dataset2.toJavaRDD().isEmpty()) {
                if (resultDataset2 == null) {
                    resultDataset2 = container.dataset2;
                } else {
                    resultDataset2 = resultDataset2.union(container.dataset2);
                }
            }

            // 对dataset3进行union操作
            if (container.dataset3 != null && !container.dataset3.toJavaRDD().isEmpty()) {
                if (resultDataset3 == null) {
                    resultDataset3 = container.dataset3;
                } else {
                    resultDataset3 = resultDataset3.union(container.dataset3);
                }
            }
        }

        DatasetContainer resultContainer = new DatasetContainer();
        resultContainer.dataset1 = resultDataset1;
        resultContainer.dataset2 = resultDataset2;
        resultContainer.dataset3 = resultDataset3;

        return resultContainer;
    }

}
